package com.young.rabbitmq.confirm;

import com.rabbitmq.client.*;
import com.young.rabbitmq.util.ConnectionUtil;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.Time;
import java.util.concurrent.TimeoutException;

public class Producer {


    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";


    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        Connection connection = ConnectionUtil.buildConnection();
        Channel channel = connection.createChannel();


        /**
         * 声明交换机
         * 参数1：交换机名称
         * 参数2：交换机类型，fanout、topic、direct、headers
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT);

        //7、声明/创建队列,这里创建了两个队列
        /*
         * 参数解析：
         * 1、队列名称
         * 2、是否定义持久化队列
         * 3、是否独占本次连接
         * 4、是否在不使用的时候自动删除
         * 5、队列其他参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1, true, false, false, null);
        channel.queueDeclare(FANOUT_QUEUE_2, true, false, false, null);

        //队列绑定交换机
        /*
            参数说明：
               参数1：队列名称
               参数2：交换机名称
               参数3：路由键
         */
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");
        channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"");
        // 8、发送消息
        /*
            参数解析
         * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
         * 参数2：路由key,简单模式可以传递队列名称
         * 参数3：消息其它属性
         * 参数4：消息内容
         */
        String routeKey = "";
        // 发泄消息之前开启发送方确认模式
        channel.confirmSelect();

        //异步监听确认和未确认的消息,为了确保程序能收到异步的回调，需要保证程序再发送完消息后不能立即结束，
        // 在本例中如果不关闭连接和通道，主线程就不会结束
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                System.out.println("未确认消息，标识：" + deliveryTag);
            }
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                System.out.println(String.format("已确认消息，标识：%d，多个消息：%b", deliveryTag, multiple));
            }
        });

        channel.addReturnListener(new ReturnListener() {
            @Override
            public void handleReturn(int replyCode, String replyText, String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.err.println("---------handle  return----------");
                System.err.println("replyCode: " + replyCode);
                System.err.println("replyText: " + replyText);
                System.err.println("exchange: " + exchange);
                System.err.println("routingKey: " + routingKey);
                System.err.println("properties: " + properties);
                System.err.println("body: " + new String(body));
            }
        });


        for (int i = 0; i < 10; i++) {
            channel.basicPublish(FANOUT_EXCHAGE,routeKey,true,null,("RabbitMQ Fanout"+i).getBytes(StandardCharsets.UTF_8));
            //Thread.sleep(1000);
            // 普通confirm模式
//            if (channel.waitForConfirms()){
//                System.out.println("消息发送成功");
//            }
        }
        //批量确认模式，直到所有信息都发布，只要有一个未确认就会IOException
//        channel.waitForConfirmsOrDie();
//        System.out.println("消息全部发送成功");



        // 9、关闭资源
       /* channel.close();
        connection.close();*/
    }
}
